#pragma once
#include <pthread.h>
#include <queue>
#include <unistd.h>
#include "sock/UdpSocket.hpp"
#include "TcpMessage.hpp"

#define MAX_THREAD_SIZE 10
#define MAX_TASK 20

class LogRegTask{
  public:
    LogRegTask(TcpSocket* NewSock, UserManager* UManager)
    :new_sock(NewSock), u_manager(UManager)
    {}

    ~LogRegTask(){
      new_sock->Close();
      delete new_sock;
    }

    bool Deal(){
      while(1){
        std::string buf;
        if(!new_sock->Recv(buf)){
          break;
        }
        TcpMessage* msg = (TcpMessage*)buf.data();
        if(!msg->Reply(new_sock, u_manager)){
          return false;
        }
      }
      return true;
    }
  private:
    TcpSocket* new_sock;
    UserManager* u_manager;
};

class ThreadPool{
  public:
    ThreadPool(int max_task = MAX_TASK, int max_thread = MAX_THREAD_SIZE)
    :IsExit(false)
    ,max_task_num(max_task)
    ,max_thread_num(max_thread){
      if(max_thread > MAX_THREAD_SIZE){
        max_thread_num = MAX_THREAD_SIZE;
      }
      cur_thread_num = max_thread_num;
      pthread_mutex_init(&pool_lock, nullptr);
      pthread_cond_init(&consumer_cond, nullptr);
    }

    ~ThreadPool(){
      pthread_mutex_destroy(&pool_lock);
      pthread_cond_destroy(&consumer_cond);
    }

    bool PushTask(LogRegTask* task){
      Lock();
      if(IsFull()){
        return false;
      }
      task_queue.push(task);
      WakeUp();
      UnLock();
      return true;
    }

    void PopTask(LogRegTask** task){
      Lock();
      while(task_queue.empty()){
        Wait();
      }
      *task = task_queue.front();
      task_queue.pop();
      UnLock();
    }

    bool PoolInit(){
      int ret;
      for(int i = 0; i < max_thread_num; ++i){
        ret = pthread_create(&consumer_pid[i], nullptr, ConsumerStart, this);
        if(IsError(ret, FATAL)){
          return false;
        }
      }
      return true;
    }

    void Quit(){
      Lock();
      IsExit = true;
      UnLock();
      while(cur_thread_num){
        WakeUpAll();
        sleep(1);
      }
      for(int i = 0; i < max_thread_num; ++i){
        pthread_join(consumer_pid[i], nullptr);
      }
    }

    static void* ConsumerStart(void* arg){
      pthread_detach(pthread_self());
      ThreadPool* tp = (ThreadPool*)arg;
      while(1){
        if(tp->IsExit){
          tp->Lock();
          --tp->cur_thread_num;
          tp->UnLock();
          pthread_exit(nullptr);
        }
        LogRegTask* task;
        tp->PopTask(&task);
        if(!task->Deal()){
          printf("Bad Require!\n");
        }
        delete task;
      }
      return nullptr;
    }
  private:
    bool IsFull(){
      return task_queue.size() == max_task_num;
    }
    
    void Lock(){
      pthread_mutex_lock(&pool_lock);
    }
    
    void UnLock(){
      pthread_mutex_unlock(&pool_lock);
    }

    void Wait(){
      pthread_cond_wait(&consumer_cond, &pool_lock);
    }

    void WakeUp(){
      pthread_cond_signal(&consumer_cond);
    }

    void WakeUpAll(){
      pthread_cond_broadcast(&consumer_cond);
    }

    bool IsExit;
    size_t max_task_num;
    int max_thread_num;
    int cur_thread_num;
    std::queue<LogRegTask*> task_queue;
    pthread_t consumer_pid[MAX_THREAD_SIZE];
    pthread_mutex_t pool_lock;
    pthread_cond_t consumer_cond;
};
